#include "dplink.h"

#define link_log(fmt, ...)  printf("[LINK]" fmt, ##__VA_ARGS__)

DPSockLink::DPSockLink()
{
    memset(&dp_param, 0, sizeof(dp_param));

    udp_fd = INVALID_SOCKET;
    tcp_fd = INVALID_SOCKET;
    unix_fd = INVALID_SOCKET;
#ifdef IO_SELECT
    nfds = INVALID_SOCKET;
#endif

    timeout             = 1000;
#ifdef IO_EPOLL
    ep_fd           = -1;
    ep_max_events   = 5;            /// current cannot be set
    ep_timeout      = 100;          /// 100ms
    ep_ev           = NULL;
#endif
    rmt_addr_len    = 0;
    memset(&rmt_addr, 0, sizeof(rmt_addr));
    memset(&rmt_addr_unix, 0, sizeof(rmt_addr_unix));

    for (int loop = 0; loop < MAX_CLIENT_NUM; loop++)
    {
        memset(&client_ctx[loop].rmt_addr, 0, sizeof(client_ctx[loop].rmt_addr));
    }

    trigger_count   = 0;

    data_len = 0;
}

int DPSockLink::get_protocol()
{
    return protocol;
}

int DPSockLink::create(DMapParam &param)
{
    //link_log("link create...\n");
    int ret;
//hi3716 does not support epoll
#ifdef IO_EPOLL
    ret = create_ev_model();
    if (ret < 0)
    {
        link_log("create epoll error, ret:%d\n", ret);
        return ret;
    }
#endif

    dp_param   = param;

    ret = create_socket();
    if (ret < 0)
    {
        return ret;
    }

    ret = pthread_create(&link_task_pid, NULL, sock_routine, this);

    return ret;
}
#ifdef IO_EPOLL
/// 创建事件模型
int DPSockLink::create_ev_model()
{
    ep_fd = epoll_create1(EPOLL_CLOEXEC);
    if (ep_fd < 0)
    {
        link_log("epoll_create1 error, errno:%d", errno);
        return -1;
    }

    ep_ev = (struct epoll_event *)malloc(sizeof(struct epoll_event) * ep_max_events);
    if (NULL == ep_ev)
    {
        return -ENOMEM;
    }

    return 0;
}
#endif

int DPSockLink::create_socket()
{
    int ret;
    struct sockaddr_in addr;
#ifdef IO_EPOLL
    struct epoll_event evt;
#endif
    if (0 != dp_param.svr_port)//udp
    {
        protocol = IPPROTO_UDP;
        addr.sin_family         = AF_INET;
        addr.sin_addr.s_addr    = dp_param.svr_addr;
        addr.sin_port           = dp_param.svr_port;
        udp_fd                  = CBB::cbb_sock_create(AF_INET, SOCK_DGRAM, 0, 1, &addr, sizeof(addr));
        if (udp_fd < 0)
        {
            link_log("creat socket error\n");
            return -1;
        }

        //link_log("creat udp sock, sock fd:%d\n", udp_fd);

        sock_svr_ctx.fd         = udp_fd;
        sock_svr_ctx.func       = svr_recv_data;
        sock_svr_ctx.context    = this;

//        CBB::cbb_socket_nonblock(udp_fd, true);
#ifdef IO_EPOLL
        evt.events              = EPOLLIN;
        evt.data.ptr            = &sock_svr_ctx;
        ret = epoll_ctl(ep_fd, EPOLL_CTL_ADD, sock_svr_ctx.fd, &evt);
        if (ret < 0)
        {
            return -1;
        }
        link_log("ip:%u create udp socket and epoll_ctl ok!\n", dp_param.svr_addr);
#elif IO_SELECT
        if (nfds < sock_svr_ctx.fd)
            nfds = sock_svr_ctx.fd;
#endif

    }
    else if (0 != dp_param.tcp_svr_port)//tcp
    {
        protocol = IPPROTO_TCP;
        addr.sin_family         = AF_INET;
        addr.sin_addr.s_addr    = dp_param.tcp_svr_addr;
        addr.sin_port           = dp_param.tcp_svr_port;
        tcp_fd                  = CBB::cbb_sock_create(AF_INET, SOCK_STREAM, 0, 1, &addr, sizeof(addr));
        if (tcp_fd < 0)
        {
            return -1;
        }
        //link_log("creat tcp sock, sock fd:%d\n", tcp_fd);
        ret  = CBB::cbb_sock_listen(tcp_fd, 10);
        if (ret < 0)
        {
            CBB::cbb_sock_close(tcp_fd);
            return -1;
        }
        //link_log("listen tcp sock ok\n");

        tcp_sock_svr_ctx.fd         = tcp_fd;
        tcp_sock_svr_ctx.func       = tcp_connection;
        tcp_sock_svr_ctx.context    = this;

#ifdef IO_EPOLL
        evt.events                  = EPOLLIN;
        evt.data.ptr                = &tcp_sock_svr_ctx;
        ret = epoll_ctl(ep_fd, EPOLL_CTL_ADD, tcp_sock_svr_ctx.fd, &evt);
        if (ret < 0)
        {
            return -1;
        }
        //link_log("ip:%u create tcp socket and epoll_ctl ok!\n", dp_param.tcp_svr_addr);

#elif IO_SELECT

    if (nfds <  tcp_sock_svr_ctx.fd)
    {
        nfds = tcp_sock_svr_ctx.fd;
    }

#endif
    }
    else if (0 != dp_param.inter_msg_addr[0])//inter unix domain
    {
        struct sockaddr_un  addr_un;
        memset(&addr_un, 0, sizeof(addr_un));

        addr_un.sun_family         = AF_UNIX;

        snprintf(addr_un.sun_path, sizeof(addr_un.sun_path), "%s", dp_param.inter_msg_addr);

        unlink(dp_param.inter_msg_addr);

        unix_fd = CBB::cbb_sock_create(AF_UNIX, SOCK_DGRAM, 0, 1, &addr_un, sizeof(addr_un));

        if (unix_fd < 0)
        {
            return -1;
        }

        sock_inter_ctx.fd       = unix_fd;
        sock_inter_ctx.func     = inter_recv_data;
        sock_inter_ctx.context  = this;
#ifdef IO_EPOLL
        evt.events              = EPOLLIN;
        evt.data.ptr            = &sock_inter_ctx;
        ret = epoll_ctl(ep_fd, EPOLL_CTL_ADD, sock_inter_ctx.fd, &evt);
        if (ret < 0)
        {
            return -1;
        }
        link_log("add:%s unix socket and epoll_ctl ok!\n", dp_param.inter_msg_addr);
#elif IO_SELECT
        if (nfds < sock_inter_ctx.fd)
            nfds = sock_inter_ctx.fd;
#endif
    }

    return  0;
}

int DPSockLink::buf_send_data(void *context, void *buf, int len, uint32_t rmt_ip, uint16_t rmt_port)
{
    DPSockLink *link = static_cast<DPSockLink *>(context);
    int ptl = link->get_protocol();
    if (ptl == IPPROTO_TCP)
    {
        int fd = link->tcp_find_socket(rmt_ip, rmt_port);
        return tcp_svr_send_data(fd, buf, len);
    }
    else if(ptl == IPPROTO_UDP)
    {
        return svr_send_data(buf, len, rmt_ip, rmt_port);
    }
    else
    {
        return -1;
    }
}

int DPSockLink::tcp_find_socket(uint32_t ip, uint16_t port)
{
    int pos;
    for(pos = 0; pos < MAX_CLIENT_NUM; pos++)
    {
        if (client_ctx[pos].client_sock_ctx.fd != 0)
        {
            if ((client_ctx[pos].rmt_addr.sin_addr.s_addr == ip)
                    && (client_ctx[pos].rmt_addr.sin_port == port))
            {
                return client_ctx[pos].client_sock_ctx.fd;
            }
        }
    }
    return -1;
}

/// 发送消息
int DPSockLink::svr_send_data(void *buf, int len, uint32_t rmt_ip, uint16_t rmt_port)
{
    struct sockaddr_in  addr;
    addr.sin_family         = AF_INET;
    addr.sin_addr.s_addr    = rmt_ip;
    addr.sin_port           = rmt_port;
    uint32_t addr_len       = sizeof(addr);

    return CBB::cbb_sendto(sock_svr_ctx.fd, buf, len, &addr, addr_len);
}

///recv data
void DPSockLink::svr_recv_data(void *ctx, uint64_t cur_time)
{
    DPSockLink     *dp_link    = static_cast<DPSockLink *>(ctx);
    if (dp_link)
    {
//        link_log("%s\n", "-----------------------------------");
        int ret = dp_link->udp_recv_data(dp_link->sock_svr_ctx.fd, dp_link->dp_param.svr_data_cb, dp_link->dp_param.svr_data_cb_ctx, cur_time);
        if (ret < 0)
        {
            /// need to restart cur link
        }
    }

}

///  recv data
int DPSockLink::udp_recv_data(int fd, FInetDataCb cb, void *cb_ctx, uint64_t cur_time)
{
//    link_log("%s, fd:%d\n"," ++++++++++++++++++++++++", fd);
    rmt_addr_len    = sizeof(rmt_addr);
    data_len = CBB::cbb_recvfrom(fd, buf, sizeof(buf), &rmt_addr, (int *)&rmt_addr_len, MSG_TRUNC);
    //link_log("recv from datalen:%d\n", data_len);
    if (data_len > 0)
    {
        if ((unsigned)data_len <= sizeof(buf))
        {
            cb(cb_ctx, buf, data_len, rmt_addr.sin_addr.s_addr, rmt_addr.sin_port, cur_time);
        }
        else
        {
            /// too larger data
        }
    }
    else if (data_len < 0)
    {
        /// need to restart linker
        return -1;
    }

    return 0;
}

/// server recv data
void DPSockLink::tcp_connection(void *ctx, uint64_t cur_time)
{
    DPSockLink     *dp_link    = static_cast<DPSockLink *>(ctx);
    dp_link->accept_new_client(cur_time);
}

void DPSockLink::accept_new_client(uint64_t cur_time)
{
    struct sockaddr_in rmt_addr;
    int fd = CBB::cbb_sock_accept(tcp_sock_svr_ctx.fd, (struct sockaddr *)&rmt_addr);
    if (fd < 0)
    {
        return;
    }
    int ret = fcntl(fd, F_SETFD, fcntl(fd, F_GETFD) | O_NONBLOCK);
    if (0 != ret)
    {
        return;
    }
    ///保存客户端连接信息
    int index = 0;
    for (; index < MAX_CLIENT_NUM; index++)
    {
        if (-1 == client_ctx[index].client_sock_ctx.fd)
        {
            break;
        }
    }

    if(index == MAX_CLIENT_NUM)
    {
        return;
    }
    /// init current client
    client_ctx[index].client_sock_ctx.fd       = fd;
    client_ctx[index].client_sock_ctx.func     = tcp_svr_recv_data;
    client_ctx[index].client_sock_ctx.context  = client_ctx + index;
    client_ctx[index].client_link       = this;
    client_ctx[index].rmt_addr          = rmt_addr;
    client_ctx[index].birth_time        = cur_time;

#ifdef IO_EPOLL
    epoll_event evt;
    evt.events   = EPOLLIN;
    evt.data.ptr = &(client_ctx[index].client_sock_ctx);
    ret = epoll_ctl(ep_fd, EPOLL_CTL_ADD, fd, &evt);
    if (ret < 0)
    {
        link_log("sock add epoll failed,ep fd:%d,fd:%d,errno:%d\n", ep_fd, fd, errno);
    }
    uint8_t *bytes = (uint8_t *)&(rmt_addr.sin_addr.s_addr);
    link_log("accept new client,index:%d,rmt addr[%u.%u.%u.%u:%u]\n", index,
                bytes[0], bytes[1], bytes[2], bytes[3], ntohs(rmt_addr.sin_port));
#elif IO_SELECT
    if (nfds < client_ctx[index].client_sock_ctx.fd)
        nfds = client_ctx[index].client_sock_ctx.fd;
#endif
}

/// 发送消息
int DPSockLink::tcp_svr_send_data(int fd, void *buf, int len)
{
    ssize_t i;
    if (!sendbuf)
    {
        sendbuf = (char *) malloc (DMAP_MAX_DATA_LEN);
        if (sendbuf == NULL)
          return -1;
        sendbuf_size = DMAP_MAX_DATA_LEN;
        sendbuf_len = 0;
    }
    /// 可用长度小于len
    if (sendbuf_size - sendbuf_len < len )
    {
        sendbuf = (char *)realloc(sendbuf, sendbuf_size + 2 *len);
        if (sendbuf == NULL)
        {
            return -1;
        }
        sendbuf_size += 2 *len;
    }
    ///empty
    if (sendbuf_len == 0 && sendbuf_size > DMAP_MAX_DATA_LEN)
    {
        free(sendbuf);
        sendbuf = (char *)malloc (DMAP_MAX_DATA_LEN);
        if (sendbuf == NULL)
        {
            return -1;
        }
        sendbuf_size = DMAP_MAX_DATA_LEN;
    }

    memcpy(sendbuf + sendbuf_len, buf, len);
    sendbuf_len += len;

   i = send(fd, sendbuf, sendbuf_len, MSG_NOSIGNAL);
   if (i < 0)
   {
        return errno;
   }
   else if (i == 0)
   {
        return errno;   /// peer maybe close,TODO:close specified TLS
   }
   else if ((int)i <= sendbuf_len)
   {
       memmove(sendbuf, sendbuf + i, sendbuf_len - i);
       sendbuf_len -= i;
   }
   return 0;
}


/// server recv data
void DPSockLink::tcp_svr_recv_data(void *ctx, uint64_t cur_time)
{
    CDPClientCtx    *client     = static_cast<CDPClientCtx *>(ctx);
    DPSockLink      *link       = client->client_link;
    int ret = link->tcp_recv_data(client->client_sock_ctx.fd, link->dp_param.tcp_svr_data_cb, link->dp_param.tcp_svr_data_cb_ctx, client->rmt_addr, cur_time);
    if (ret < 0)
    {
        /// need to restart cur link
       //link_log("server fd recv data fialed:%d\n", client->client_sock_ctx.fd);
    }
}

int DPSockLink::tcp_recv_data(int fd, FTcpDataCb cb, void *cb_ctx, struct sockaddr_in &addr, uint64_t cur_time)
{

    //for tmp
    (void)cb;
    (void)cb_ctx;
    (void)cur_time;

   rmt_addr = addr;
   int readn;
   if (!recvbuf)
   {
       recvbuf = (char *) malloc (DMAP_MAX_DATA_LEN);
       if (recvbuf == NULL)
         return -1;
       bufsize = DMAP_MAX_DATA_LEN;
       buflen = 0;
   }
   /// full
   if (bufsize - buflen <= 0 && bufsize < 4194304)/// 限制在4M内
   {
       recvbuf = (char *)realloc(recvbuf, bufsize + 1000);
       if (recvbuf == NULL)
       {
           return -1;
       }
       bufsize = bufsize +1000;
   }
   ///empty
   if (buflen == 0 && bufsize > DMAP_MAX_DATA_LEN)
   {
       free(recvbuf);
       recvbuf = (char *)malloc (DMAP_MAX_DATA_LEN);
       if (recvbuf == NULL)
       {
           return -1;
       }
       bufsize = DMAP_MAX_DATA_LEN;
   }

    readn = (int) recv(fd, recvbuf + buflen, (int) (bufsize - buflen), 0);
    if (readn == 0)
    {
        return errno;
    }
    else if (readn < 0)
    {
        return errno;
    }
    else
    {
//        link_log("recv data:%s\n", recvbuf);


        //something to do

//     cb(cb_ctx, recvbuf, buflen, rmt_addr.sin_addr.s_addr, rmt_addr.sin_port, cur_time);
        return 0;
    }
}

/// recv data
void DPSockLink::inter_recv_data(void *ctx,/* struct epoll_event *ev, */uint64_t cur_time)
{
    DPSockLink     *dp_link    = static_cast<DPSockLink *>(ctx);

    int ret = dp_link->udp_unix_recv_data(dp_link->sock_inter_ctx.fd, dp_link->dp_param.inter_msg_cb, dp_link->dp_param.inter_msg_cb_ctx, cur_time);
    if (ret < 0)
    {
        /// need to restart cur link
    }
}

/// unix recv data
int DPSockLink::udp_unix_recv_data(int fd, FUnixDataCb cb, void *cb_ctx, uint64_t cur_time)
{
    data_len = CBB::cbb_recvfrom(fd, buf, sizeof(buf), &rmt_addr_unix, (int *)&rmt_addr_len, MSG_TRUNC);
    if (data_len > 0)
    {
        if ((unsigned)data_len <= sizeof(buf))
        {
            cb(cb_ctx, buf, data_len, rmt_addr_unix.sun_path, cur_time);
        }
        else
        {
            /// too larger data
           link_log("server unix fd recv too larger data,len:%d\n", data_len);
        }
    }
    else if (data_len < 0)
    {
        /// need to restart linker
        return -1;
    }

    return 0;
}

/*
 * epoll task
 */
void *DPSockLink::sock_routine(void *arg)
{    
    uint64_t time = CBB::gettime_ms();
    DPSockLink *sock_link = (DPSockLink *)arg;       // cur link
    //link_log("link:%p running\n", sock_link);
#ifdef IO_EPOLL
    int ret;
    struct epoll_event *ev = sock_link->ep_ev;

    sock_link->link_task_exit = false;

    ret = prctl(PR_SET_NAME, "dp_link", NULL, NULL, NULL);
    if (ret < 0)
    {
        link_log("prctl error\n");
    }

    //start service
    link_log("link:%p->%d running\n", sock_link, sock_link->ep_fd);
    while (true)
    {
        ret = epoll_wait(sock_link->ep_fd, ev, sock_link->ep_max_events, sock_link->ep_timeout);

        if (ret < 0)
        {
            if (EINTR == errno)
            {
                continue;
            }

         link_log("link:%p->%d epoll wait failed, task to exit, errno:%d\n", sock_link, sock_link->ep_fd, errno);
            break;
        }
        else if (ret > 0)
        {
            // deal event
            sock_link->sock_deal_events(ret, time);
        }

        // 处理触发器
        sock_link->sock_deal_trigger(time);
    }

//    link_log("link:%p->%d task exit now\n", sock_link, sock_link->ep_fd);
//    sock_link->link_task_exit = true;

#elif IO_SELECT
    while (true)
    {
        //select
        sock_link->deal_select(time);
    }



#endif
    return NULL;
}

#ifdef IO_SELECT
void DPSockLink::deal_select(uint64_t time)
{
    int ret;
    static fd_set read_set, write_set, error_set;
    FD_ZERO(&read_set);
    FD_ZERO(&write_set);
    FD_ZERO(&error_set);

    if (INVALID_SOCKET != sock_svr_ctx.fd)//udp
    {
        FD_SET(sock_svr_ctx.fd, &read_set);
    }

    if (INVALID_SOCKET != sock_inter_ctx.fd)//unix domain
    {
        FD_SET(sock_inter_ctx.fd, &read_set);
    }

    if (INVALID_SOCKET != tcp_sock_svr_ctx.fd)//tcp listen
    {
        FD_SET(tcp_sock_svr_ctx.fd, &read_set);
    }

    for (int i = 0; i < MAX_CLIENT_NUM; i++)//tcp client
    {
        if (client_ctx[i].birth_time  != 0 && client_ctx[i].client_sock_ctx.fd != INVALID_SOCKET)
        {
            FD_SET(client_ctx[i].client_sock_ctx.fd, &read_set);
        }
    }

    struct timeval tv;
    tv.tv_sec   = timeout / 1000;
    tv.tv_usec  = 1000 * (timeout % 1000);
    ret = select(nfds + 1, &read_set, NULL, NULL, &tv);
    if (ret < 0)
    {
        link_log("select failed\n");
        return;
    }

//    link_log("select result:%d, nfds:%d, sock_svr_ctx.fd:%d\n", ret, nfds, sock_svr_ctx.fd);

    if ( (INVALID_SOCKET != sock_svr_ctx.fd) && FD_ISSET(sock_svr_ctx.fd, &read_set) )
    {
        //link_log("select...fd:%d\n", sock_svr_ctx.fd);
        sock_svr_ctx.func(sock_svr_ctx.context, time);
//        udp_recv_data(sock_svr_ctx.fd, dp_param.svr_data_cb, dp_param.svr_data_cb_ctx, time);
//        CBB::cbb_recvfrom(sock_svr_ctx.fd, buf, sizeof(buf), &rmt_addr, (int *)&rmt_addr_len, 0);//MSG_TRUNC

    }

    if ((INVALID_SOCKET != sock_inter_ctx.fd) && FD_ISSET(sock_inter_ctx.fd, &read_set))
    {
        sock_inter_ctx.func(sock_inter_ctx.context, time);
//        udp_unix_recv_data(sock_inter_ctx.fd, dp_param.inter_msg_cb, dp_param.inter_msg_cb_ctx, time);
    }

    if (INVALID_SOCKET != tcp_sock_svr_ctx.fd && FD_ISSET(tcp_sock_svr_ctx.fd, &read_set))
    {
        link_log("tcp accept\n");
        tcp_sock_svr_ctx.func(tcp_sock_svr_ctx.context, time);
    }

    for (int loop = 0; loop < MAX_CLIENT_NUM; loop++)
    {
        if (INVALID_SOCKET != client_ctx[loop].client_sock_ctx.fd && FD_ISSET(client_ctx[loop].client_sock_ctx.fd, &read_set))
        {
            link_log("tcp client %d , fd:%d\n", loop, client_ctx[loop].client_sock_ctx.fd);
            client_ctx[loop].client_sock_ctx.func(client_ctx[loop].client_sock_ctx.context, time);
        }
    }

}
#endif

#ifdef IO_EPOLL
/*
 * deal epoll events
 */
void DPSockLink::sock_deal_events(int num, uint64_t time)
{
    int i;
    struct epoll_event ev;
    TSockEpollCtx *ctx;

    // deal come data
    for (i = 0; i < num; i++)
    {
        ev = *(ep_ev + i);           // copy avoid mangle
        if (NULL != ev.data.ptr)
        {
            ctx = (TSockEpollCtx *)ev.data.ptr;
            /// 这里没有错误处理情况,需要注意,用户处理函数需要处理错误事件
            ctx->func(ctx->context, time);// &ev,
        }
        else
        {
          link_log("fatal error happened, epoll event with no user\n");
        }
    }
}

/*
 * deal epoll trigger
 */
void DPSockLink::sock_deal_trigger(uint64_t time)
{
    if (dp_param.event_trigger_cb)
    {
        dp_param.event_trigger_cb(dp_param.ev_trigger_ctx, time);
    }
}

#endif


